/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.LongValueCodec;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.reading.IDrawerIndexReader.CardInfo;

public class DeltaEncodedAscendingUnsignedLongCardReader extends
    AbstractCardReader<LongType> {
  private static final Logger LOG = 
      Logger.getLogger(DeltaEncodedAscendingUnsignedLongCardReader.class.getName());

  private enum ReaderState {
    NeedsCard,
    Reading
  }

  private long cardMaxIndex;
  private long cardMaxValue;
  private int cardOffsetLimit;
  private long currentValue;
  private final ObjectLongReference decoded;
  private final VariableLengthUnsignedLongSequenceCodec longCodec;
  private final LongValueCodec longManip;
  private ReaderState readerState;

  public DeltaEncodedAscendingUnsignedLongCardReader() {
    super(LongTools.INSTANCE);
    this.longCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;
    this.longManip = LongTools.INSTANCE.getGenericCodec();
    this.decoded = new ObjectLongReference();
    this.readerState = ReaderState.NeedsCard;
  }

  @Override
  public void jumpCursor(final int targetEntryIndex,
                         final IValueReference<LongType> valueOutput) {
    assert this.readerState == ReaderState.Reading;
    // Check condition that request must be on this card
    assert targetEntryIndex >= this.cardInfo.getFirstEntryIndex();
    assert targetEntryIndex < this.cardMaxIndex;
    assert targetEntryIndex >= this.current.index;

    while (this.current.index <= targetEntryIndex) {
      this.stepCursorInternal();
    }

    this.longManip.setLongValue(this.currentValue, valueOutput);
  }

  @Override
  public boolean
      jumpCursorToValue(final IValueReference<LongType> minimumTargetValue,
                        final IValueReference<LongType> valueOut) {
    long targetValue = this.longManip.getLongValue(minimumTargetValue);
    
    if (targetValue > this.cardMaxValue) {
      this.readerState = ReaderState.NeedsCard;
      return false;
    } else if (targetValue == this.cardMaxValue) {
      this.readerState = ReaderState.NeedsCard;
      this.longManip.setLongValue(cardMaxValue, valueOut);
      this.current.index = this.cardMaxIndex + 1;
      return true;
    }

    while (this.current.index < this.cardMaxIndex && this.currentValue < targetValue) {
      this.stepCursorInternal();
    }
    
    if (this.current.index <= this.cardMaxIndex) {
      this.longManip.setLongValue(this.currentValue, valueOut);
      return true;
    } else if (this.current.index == (this.cardMaxIndex + 1)) {
      // Since we always leave the current index one ahead of the current value
      this.readerState = ReaderState.NeedsCard;
      this.longManip.setLongValue(cardMaxValue, valueOut);
      this.current.index = this.cardMaxIndex + 1;
      return true;
    } else {
      return false;
    }
  }

  @Override
  public void resetCard(final CardInfo<LongType> cardInfo) {
    super.resetCard(cardInfo);
    this.cardOffsetLimit = this.current.byteOffset + ColumnStore.BYTES_IN_CARD;
    this.cardMaxIndex =
        this.cardInfo.getFirstEntryIndex() + this.cardInfo.getEntryCount() - 1;
    this.cardMaxValue = this.drawerBuffer.getLong(this.current.byteOffset);
    
//    if (this.cardMaxValue != this.longManip.getLongValue(cardInfo.getLastValue())) {
//      String format = "cardMaxValue=%s != cardInfo.getLastValue()=%s.";
//      LOG.warning(String.format(format, 
//                                this.cardMaxValue, 
//                                this.longManip.getLongValue(cardInfo.getLastValue())));
//    } 
    
    this.longManip.setLongValue(this.cardMaxValue, cardInfo.getLastValue());
    this.current.byteOffset += 8; // adjust for max value at start of card
    this.currentValue = 0;
    this.readerState = ReaderState.Reading;

    assert this.cardMaxIndex > 0;
  }

  @Override
  public boolean stepCursor(final IValueReference<LongType> valueOut) {
    if (this.readerState == ReaderState.NeedsCard) { return false; }
    assert this.readerState == ReaderState.Reading;

    if (this.current.index == this.cardMaxIndex) {
      this.longManip.setLongValue(this.cardMaxValue, valueOut);
      this.readerState = ReaderState.NeedsCard;
      this.current.index++; // So that CardReader.getCurrentIndex() still works.
      return true;
    }

    this.stepCursorInternal();
    this.longManip.setLongValue(this.currentValue, valueOut);
    return true;
  }

  private void stepCursorInternal() {
    this.current.index++;

    int bytesConsumed =
        this.longCodec.decode(this.drawerBuffer, this.decoded,
                              this.current.byteOffset, this.cardOffsetLimit);
    assert bytesConsumed > 0;
    assert this.decoded.getValue() >= 0;

    this.current.byteOffset += bytesConsumed;
    this.currentValue += this.decoded.getValue();
  }

}
